Streaming Output Modes & Triggers

When working with Structured Streaming, you must specify how and when Spark writes computed results to the output sink. This is controlled by Output Modes and Streaming Triggers.

Streaming Output Modes

The Output Mode defines which part of your streaming DataFrame is written to the sink during each micro-batch trigger:

Streaming Triggers (When to Process)

A Trigger defines the timing of stream evaluations:

# Configure trigger interval
df.writeStream.trigger(processingTime="10 seconds").start()

• Micro-Batch (Default): If no trigger is specified, Spark runs micro-batches as fast as possible (starts the next one immediately after the previous one finishes).
• Processing Time: Runs the streaming query at regular clock intervals (e.g. every 10 seconds, 1 minute).
• Available Now: Processes all available data in the source, writes it, and stops the streaming query automatically (ideal for cost-effective daily/hourly batch-like streaming runs).
• Continuous Processing: Experimental low-latency engine that processes events record-by-record, achieving sub-millisecond latencies (restricted to simple maps and select transformations).

PySpark Code Example: Aggregations in Complete Mode

Here is a complete script demonstrating event rate aggregations using Complete Mode and processing-time triggers:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("Streaming Output Modes") \
    .master("local[*]") \
    .getOrCreate()

# 2. Ingest Rate Stream (Generates 'timestamp' and 'value' columns automatically)
rate_stream_df = spark.readStream \
    .format("rate") \
    .option("rowsPerSecond", 5) \
    .load()

# 3. Apply Grouping Aggregations (Even/Odd value counts)
aggregated_df = rate_stream_df.withColumn("type", F.when(F.col("value") % 2 == 0, "Even").otherwise("Odd")) \
                              .groupBy("type") \
                              .count()

# 4. Write stream using COMPLETE mode
# This aggregates and updates the complete count table in console every 5 seconds
query = aggregated_df.writeStream \
    .format("console") \
    .outputMode("complete") \
    .trigger(processingTime="5 seconds") \
    .option("checkpointLocation", "complete_checkpoints") \
    .start()

query.awaitTermination()